package com.hery.driver;

import com.hery.functions.AlertManagerSinkFunction;
import com.hery.functions.RuleMatchBroadCastProcessFunction;
import com.hery.functions.SinkFunctions;
import com.hery.functions.SourceFunctions;
import com.hery.pojo.KafkaAlertLog;
import com.hery.pojo.Rule;
import com.hery.utils.Config;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 车机端埋点日志告警，用 {@link Driver}调用
 */
public class JsonAlertJobDriver {
    public static void main(String[] args) throws Exception {

        // 全局配置
        Configuration config = Config.initApplicationConfig(args, JsonAlertJobDriver.class);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);

        env.setParallelism(2);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(4,3000L));
        //env.setStateBackend(new FsStateBackend("hdfs:///flink/checkpoint/cp/flink-real-time-alert/cp"));
        env.setStateBackend(new FsStateBackend("file:///qisi/gitlab/flink-real-time-alert/cp"));
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.getCheckpointConfig().disableCheckpointing();

        // 捕获 MySQL 中的规则流
        SingleOutputStreamOperator<Rule> mysqlRuleStream = env.addSource(SourceFunctions.getCustomMysqlSource(config)).name("mysqlRuleStream");

        // 获取 Kafka 日志流
        SingleOutputStreamOperator<String> kafkaSource = env.addSource(SourceFunctions.getKafkaEventSource()).name("kafkaSourceStream");

        // 连接 日志流 & 规则广播流
        ConnectedStreams<String, Rule> connectStream = kafkaSource.connect(mysqlRuleStream.broadcast());
        // 匹配出告警的 LogMessage
        SingleOutputStreamOperator<KafkaAlertLog> alertStream = connectStream.process(new RuleMatchBroadCastProcessFunction(config));

        // 发送告警信息到 AlertManager
        alertStream.addSink(new AlertManagerSinkFunction(config));

        // 告警服务发送到 Mysql 满足告警统计的需求
        alertStream.addSink(SinkFunctions.getMysqlAlertSinkFunction(config));

        // 告警结果发送到 Kafka 满足监控的需求
        alertStream.filter(KafkaAlertLog::isSentryAlertData).map(KafkaAlertLog::getJson).addSink(SinkFunctions.getFlinkKafkaProducer());

        alertStream.print();

        env.execute(JsonAlertJobDriver.class.getSimpleName());
    }

}
